package com.itheima.rabbitmq.producer.config;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.stereotype.Component;

import java.util.Date;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Semaphore;

/**
 * GY
 */
@Component
public class MessageSender {
    @Autowired
    private ApplicationContext context;

    // TODO 需要定时任务做清理 遍历根据发送时间判断 key中可携带相关时间信息(或针对Value对象进一步封装)
    // 这个属性，肯定也是只有这一个对象
    private Map<String, Semaphore> semaphoreMap = new ConcurrentHashMap<>(1024);

    @Autowired
    private RabbitTemplate rabbitTemplate;

    private Logger logger = LoggerFactory.getLogger(MessageSender.class);

    public void sendMQMessage(String exchange, String routingKey, String message) {

        // 设置一个id
        String correlationDataId = UUID.randomUUID().toString();
        CorrelationData correlationData = new CorrelationData(correlationDataId);

        System.out.println("send id = " + correlationData.getId());
        rabbitTemplate.convertAndSend(exchange, routingKey, message, correlationData);

        // 构建信号量，用于异步消息落库完成，confirm回调线程同步操作
        // 每个线程在执行下面这行代码之后，都会创建一个属于线程自己的对象
        Semaphore semaphore = new Semaphore(0); // 一百个线程执行进来就会生成100个Semaphore

        // 把对象统一放置到一个地方进行 映射存储k-v
        // 为了保证线程安全访问，用ConcurrentHashMap
        semaphoreMap.put(correlationDataId, semaphore);

        logger.info("send message success : {}", message);

        // 异步落库
        MQMessage mqMessage = new MQMessage(correlationDataId, exchange, routingKey, message);
        new MessageDBTask(mqMessage, semaphore, this.context).executor();

    }

    /**
     * 用于存储到mysql中的消息记录实体
     */
    public static class MQMessage {
        private String correlationDataId;
        private String exchangeName;
        private String routingKey;
        private String message; // 消息内容
        private Date sendTime = new Date(); // 消息发送时间
        private int reTimes = 0; // 已经重试次数
        private int status = 0;  // 默认发送失败 0是失败，1是成功

        public MQMessage(String correlationDataId,String exchangeName, String routingKey, String message) {
            this.exchangeName = exchangeName;
            this.routingKey = routingKey;
            this.message = message;
            this.correlationDataId = correlationDataId;
        }

        @Override
        public String toString() {
            final StringBuilder sb = new StringBuilder("MQMessage{");
            sb.append("correlationDataId='").append(correlationDataId).append('\'');
            sb.append(", exchangeName='").append(exchangeName).append('\'');
            sb.append(", routingKey='").append(routingKey).append('\'');
            sb.append(", message='").append(message).append('\'');
            sb.append(", sendTime=").append(sendTime);
            sb.append(", reTimes=").append(reTimes);
            sb.append(", status=").append(status);
            sb.append('}');
            return sb.toString();
        }
    }

    public Semaphore getSemphone(String correlationDataId) {
        return semaphoreMap.get(correlationDataId);
    }

    public void removeSemphone(String correlationDataId){
        this.semaphoreMap.remove(correlationDataId);
    }

}
